package com.chukun.flink.stream.helloword;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Objects;

/**
 * @author chukun
 * @version 1.0.0
 * @description 单词计数，flink helloworld
 * @createTime 2022年05月09日 00:05:00
 */
public class WordCount {

    public static final String [] WORDS = new String[]{
            "com.chukun.flink.stream.helloword.flink.start",
            "com.chukun.flink.stream.helloword.flink.fun",
            "com.chukun.flink.stream.helloword.flink.word.count",
            "com.chukun.flink.stream.helloword.flink.stream",
    };

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> wordText = env.fromElements(WORDS);

        // map操作
        DataStream<Tuple2<String, Integer>> mappedWordStream = wordText.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (text, collector) -> {
            String[] tokens = text.toLowerCase().split("\\.");
            for (String token : tokens) {
                if (!Objects.equals(token, "") || token.length() > 0) {
                    collector.collect(new Tuple2<>(token, 1));
                }
            }
        }).returns(TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {})); // 强制指定返回类型

        // reduce操作
        DataStream<Tuple2<String, Integer>> counts = mappedWordStream.keyBy((KeySelector<Tuple2<String, Integer>, String>) tuple2 -> tuple2.f0)

                .sum(1);

        counts.print("hello datastream");

        env.execute();
    }
}
